package com.corn.test.zk;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.junit.Before;
import org.junit.Test;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2022/7/11 15:17
 */
@Slf4j
public class ZkClientDemo {

    private static String zkServers = "node1:2181";

    private static int sessionTimeout = 2000;

    private static ZooKeeper zkClient;

    @Before
    public void init() throws Exception {
        zkClient = new ZooKeeper(zkServers, sessionTimeout, new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                log.info("事件类型:{} ,路径:{} ,状态:{} ", watchedEvent.getType(), watchedEvent.getPath(), watchedEvent.getState());
            }
        });
    }

    /**
     * 创建子节点
     *
     * @throws KeeperException
     * @throws InterruptedException
     */
    @Test
    public void testCreateNode() throws KeeperException, InterruptedException {
        /**
         * String path, 节点路径
         * byte[] data, 数据
         * List<ACL> acl, 访问权限 OPEN_ACL_UNSAFE所有人全部权限
         * CreateMode createMode 节点类型 PERSISTENT持久化节点
         */
        zkClient.create("/demo", "demoData".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    /**
     * 获取子节点目录并监控
     */
    @Test
    public void testWatchNode() throws KeeperException, InterruptedException {
        List<String> children = zkClient.getChildren("/demo", true);
        log.info(String.join(",", children));
        Thread.sleep(600 * 1000);
    }

    @Test
    public void testGetData() throws KeeperException, InterruptedException {
        Stat stat = new Stat();

        byte[] data = zkClient.getData("/demo", new Watcher() {
            @SneakyThrows
            @Override
            public void process(WatchedEvent watchedEvent) {
                log.info("getData 事件类型:{} ,路径:{} ,状态:{} ", watchedEvent.getType(), watchedEvent.getPath(), watchedEvent.getState());
                byte[] changeData = zkClient.getData("/demo", true, null);
                log.info("节点变化数据:{}", new String(changeData));
            }
        }, stat);
        log.info("节点数据:{}", new String(data));
        Thread.sleep(600 * 1000);
    }


    /**
     * 查看节点是否存在
     *
     * @throws Exception
     */
    @Test
    public void testExist() throws Exception {
        Stat exists = zkClient.exists("/demo", false);
        log.info(exists.toString());
    }
}
